package com.shujia.flink.core

import org.apache.flink.api.common.RuntimeExecutionMode
import org.apache.flink.streaming.api.scala._

object Demo2BatchWC {
  def main(args: Array[String]): Unit = {
    /**
     * 创建FLink环境
     *
     */
    //获取一个环境，flink在内会自动判断是本地还是集群
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    /**
     * 修改flink处理模式
     * 流处理-一条一条处理，每次处理都会输出一个结果
     * 批处理：一起处理，输出最终结果, 批处理模式不能用于处理无界流的数据
     *
     * 有界流：比如读取文件，又开始有结束
     * 无界流：比如socket,有开始没有结束
     *
     *
     * flink流批统一：同一套代码既能做流处理也能做批处理
     *
     *
     * 当使用批处理模式时，flink底层处理数据的方式和spark一样，先执行上游task再执行下游task
     *
     */
    //默认是流处理模式
    env.setRuntimeMode(RuntimeExecutionMode.BATCH)


    /**
     * 使用flink处理文件
     *
     */

    //读取文件---- 有界流
    val linesDS: DataStream[String] = env.readTextFile("data/words.txt")

    //1、将一行转换成多行
    val wordsDS: DataStream[String] = linesDS.flatMap(line => line.split(","))

    //2、转换成kv格式
    val kvDS: DataStream[(String, Int)] = wordsDS.map(word => (word, 1))

    //3、安装单词分组
    val keyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(kv => kv._1)

    //4、统计单词的数量，指定对哪一个列求和，指定下标，下标从0开始
    val countDS: DataStream[(String, Int)] = keyByDS.sum(1)

    countDS.print()

    env.execute()

  }

}
